package com.example.until;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * ClassName: KafkaUtils
 * PackageName: com.example.until
 * Describe:
 *
 * @author: Cheng Wenying
 * @Create: 2023/11/1
 * @version:
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaUtils {
    private final KafkaTemplate<String, String> kafkaTemplate;


    /**
     * 发送消息
     */
    public void kafkaSendMsg(String topicName, String msg) {
        kafkaTemplate.send(topicName, msg);
        log.info("kafka成功发送消息给：" + topicName + "，内容为：" + msg);
    }

    /**
     * 监听消息
     */
    @KafkaListener(topics = {"test"}, groupId = "java-group", containerFactory = "kafkaListenerContainerFactory")
    public void kafkaListener(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {

        log.info("这是消费者在消费消息：" + record.topic() + "----" + record.partition() + "----" + record.value());

        ack.acknowledge();
    }
}
